/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.common;

import java.io.File;

import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.io.FileReference;
import edu.uci.ics.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
import edu.uci.ics.hyracks.data.std.primitive.IntegerPointable;
import edu.uci.ics.hyracks.data.std.primitive.UTF8StringPointable;
import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
import edu.uci.ics.hyracks.storage.am.common.api.IFreePageManager;
import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
import edu.uci.ics.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
import edu.uci.ics.hyracks.storage.am.common.freepage.LinkedListFreePageManager;
import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
import edu.uci.ics.hyracks.storage.am.common.tuples.TypeAwareTupleWriterFactory;
import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
import edu.uci.ics.hyracks.storage.common.file.IFileMapManager;
import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;

public class HyracksBtreeLexicon {

  protected BTree lexicon;

  public HyracksBtreeLexicon() {
    super();
  }

  public void close() {
    this.lexicon.close();
  }

  public void initialize(final IBufferCache bufferCache, final String filename,
                         final IFileMapProvider fileMapProvider,
                         final boolean createNew) throws HyracksDataException {

    // Initialize lexicon file
    FileReference btreeFile = new FileReference(new File(filename));
    int btreeFileId = -1;
    if (createNew) {
      assert !fileMapProvider.isMapped(btreeFile);
      bufferCache.createFile(btreeFile);
      btreeFileId = fileMapProvider.lookupFileId(btreeFile);
    } else {
      if (!fileMapProvider.isMapped(btreeFile)) {
        IFileMapManager fileMap = (IFileMapManager) fileMapProvider;
        fileMap.registerFile(btreeFile);
      }
      btreeFileId = fileMapProvider.lookupFileId(btreeFile);
    }
    bufferCache.openFile(btreeFileId);

    // Declare lexicon fields
    int fieldCount = 2;
    ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
    // token (key)
    typeTraits[0] = UTF8StringPointable.TYPE_TRAITS;
    // headerPage
    typeTraits[1] = IntegerPointable.TYPE_TRAITS;

    // Declare lexicon key
    int keyFieldCount = 1;
    IBinaryComparator[] cmps = new IBinaryComparator[keyFieldCount];
    cmps[0] =
        PointableBinaryComparatorFactory.of(UTF8StringPointable.FACTORY)
            .createBinaryComparator();
    MultiComparator btreeCmp = new MultiComparator(cmps);

    // Initialize various btree object factories
    TypeAwareTupleWriterFactory tupleWriterFactory =
        new TypeAwareTupleWriterFactory(typeTraits);
    ITreeIndexFrameFactory leafFrameFactory =
        new BTreeNSMLeafFrameFactory(tupleWriterFactory);
    ITreeIndexFrameFactory interiorFrameFactory =
        new BTreeNSMInteriorFrameFactory(tupleWriterFactory);
    ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();

    IFreePageManager freePageManager =
        new LinkedListFreePageManager(bufferCache, btreeFileId, 0, metaFrameFactory);

    this.lexicon =
        new BTree(bufferCache, typeTraits.length, btreeCmp, freePageManager,
                  interiorFrameFactory, leafFrameFactory);

    if (createNew) {
      try {
        this.lexicon.create(btreeFileId);
      } catch (Exception e) {
        throw new HyracksDataException(e);
      }
    }

    this.lexicon.open(btreeFileId);
  }

}